﻿#region License
/* ---------------------------------------------------------------- *
 *
 * This file is part of the Xcoordination Application Space
 * ("XcoAppSpace") http://www.xcoordination.com
 *
 * Copyright (C) 2009 Xcoordination http://www.xcoordination.com
 *
 * XcoAppSpace is free software; you can redistribute it and/or
 * modify it under the terms of version 2 of the GNU General
 * Public License as published by the Free Software Foundation.
 *
 * XcoAppSpace is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses/
 * or write to the Free Software Foundation, Inc., 51 Franklin
 * Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * ---------------------------------------------------------------- */
#endregion

using System;
using System.Collections.Generic;
using Microsoft.Ccr.Core;
using XcoAppSpaces.Contracts.CcrExtensions;
using XcoAppSpaces.Contracts.Messaging;
using XcoAppSpaces.Contracts.Worker;
using System.Threading;
using XcoAppSpaces.Contracts.Logging;
using XcoAppSpaces.Contracts.Security;
using System.Net;
using XcoAppSpaces.Contracts.WorkerResolve;

namespace XcoAppSpaces.Core.Communication.WorkerResolve
{
	class ResolveWorker : ResolveWorkerContract, IDisposable
	{
		private class WaitInfo : IDisposable
		{
			private readonly Guid requestId;
			private readonly Dictionary<Guid, WaitInfo> responses;

			public ManualResetEvent Event { get; private set; }
			public ResolveResponse Response { get; set; }

			public WaitInfo(Guid requestId, Dictionary<Guid, WaitInfo> responses)
			{
				this.requestId = requestId;
				this.responses = responses;
				lock (responses)
				{
					responses.Add(requestId, this);
				}

				Event = new ManualResetEvent(false);
				Response = null;
			}

			#region IDisposable Members

			public void Dispose()
			{
				lock (responses)
				{
					responses.Remove(requestId);
				}
			}

			#endregion
		}

		private readonly IXcoWorkerRegistry localWorkers;
		private readonly IXcoSecurityService securityService;
		private readonly Handler<MessageToSend> messageProcessor;
		private readonly DispatcherQueue dpq;
		private readonly CommunicationEvents commEvents;
		private readonly Dictionary<Guid, WaitInfo> responses = new Dictionary<Guid, WaitInfo>();
		private readonly Port<ResolveResponse> responsePort = new Port<ResolveResponse>();
		private readonly IXcoLogger log = null;
		private readonly string resolveWorkerIdentifier;
		
		private const int WaitTimeout = 10000;

		public ResolveWorker(IXcoWorkerRegistry localWorkers, IXcoSecurityService securityService, Handler<MessageToSend> messageProcessor, DispatcherQueue dpq, CommunicationEvents commEvents)
		{
			this.localWorkers = localWorkers;
			this.securityService = securityService;
			this.messageProcessor = messageProcessor;
			this.dpq = dpq;
			this.commEvents = commEvents;
			this.log = XcoLoggerFactory.CreateInstance(this.GetType());
			resolveWorkerIdentifier = localWorkers.GenerateIdentifier(typeof(ResolveWorkerContract), null);
			Arbiter.Activate(dpq,
				Arbiter.Receive(true, this, ProcessResolveRequest),
				Arbiter.Receive(true, responsePort, ProcessResolveResponse));
		}

		private void ProcessResolveRequest(ResolveRequest request)
		{
			ResolveStatus status;
			Guid sessionID = Guid.Empty;
			XcoWorkerInfo worker = null;
		    string address = request.ResponsePort.GetPortInfo().RemoteAddress;
			if (localWorkers.Exists(request.WorkerIdentifier, true, true))
			{
				worker = localWorkers.Resolve(request.WorkerIdentifier, true);
				XcoAuthenticationResult auth = securityService.Authenticate(address, request.ClientCredentials);
				if (!auth.IsAuthenticated)
				{
					status = ResolveStatus.NotAuthenticated;
				}
				else
				{
					sessionID = auth.SessionID;
					if (securityService.Authorize(address, auth.SessionID, worker.Worker))
						status = ResolveStatus.Ok;
					else
						status = ResolveStatus.NotAuthorized;
				}
			}
			else
			{
				status = ResolveStatus.WorkerNotFound;
			}
			log.Debug("ResolveWorker: ResolveRequest received from " + address + " for: " + request.WorkerIdentifier + " --> " + status);
			commEvents.RaiseOnWorkerResolvedEvent(request.WorkerIdentifier, address, status);
			request.ResponsePort.Post(new ResolveResponse { RequestID = request.ID, Status = status, SessionID = sessionID, ProcessorAttributes = (worker != null ? worker.ProcessorAttributes : null)});
		}

		private void ProcessResolveResponse(ResolveResponse response)
		{
			//Console.WriteLine("ResolveResponse received.");
			//find which request the given response belongs to
			lock (responses)
			{
				Guid reqKey = response.RequestID;
				if (responses.ContainsKey(reqKey))
				{
					responses[reqKey].Response = response;
					responses[reqKey].Event.Set();
				}
			}
		}

		public ResolveResponse Resolve(string workerAddress, string workerIdentifier, string commServiceName, ICredentials credentials)
		{
			log.Debug("ResolveWorker: Resolving Worker: Address = " + workerAddress + ", Identifier = " + workerIdentifier);
			//send request to remote space
			var req = new ResolveRequest
			          	{
			          		WorkerAddress = workerAddress,
							WorkerIdentifier = workerIdentifier,
			          		ResponsePort = responsePort,
			          		ID = Guid.NewGuid(),
			          		ClientCredentials = securityService.GetClientCredentials(credentials, workerAddress)
			          	};
			using (var waitInfo = new WaitInfo(req.ID, responses))
			{
				bool communicationError = false;
				Exception error = null;
				var exceptionPort = new Port<Exception>();
				Arbiter.Activate(dpq, Arbiter.Receive(true, exceptionPort, ex => { waitInfo.Event.Set(); communicationError = true; error = ex; }));
				
				messageProcessor(new MessageToSend(workerAddress, commServiceName,
					new CausalityContext(new CausalityInfo { Name = "cResolve", ExceptionPort = exceptionPort }),
					new WorkerMessage(resolveWorkerIdentifier, Guid.Empty, req), null, null));
				
				//wait for response
				waitInfo.Event.WaitOne(WaitTimeout);
				
				log.Debug("ResolveWorker: Resolve finished --> " + (waitInfo.Response != null ? waitInfo.Response.Status.ToString() : (communicationError ? "communication error!" : "no response!")));
				if (waitInfo.Response != null)
					return waitInfo.Response;

			    return new ResolveResponse { Status = communicationError ? ResolveStatus.CommunicationError : ResolveStatus.NoResponse, Error = error };
			}
		}

		public void Dispose()
		{
			//wake up all waiting requests
			//Console.WriteLine("ResolveWorker: Dispose called.");
			lock (responses)
			{
				foreach (WaitInfo waiter in responses.Values)
				{
					waiter.Event.Set();
				}
			}
		}
	}
}
